package com.zhang.hadoop.flink.test10;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author: zhang yufei
 * @createTime:2022/11/12 11:52
 * @description:
 */
public class LoginFailDetectExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.获取登录数据流
        SingleOutputStreamOperator<LoginEvent> loginEventStream = env.fromElements(
                new LoginEvent("ouyanghui", "yindao", "fail", 2000L),
                new LoginEvent("ouyanghui", "gangmen", "fail", 3000L),
                new LoginEvent("ouyanghui", "gangmen", "fail", 4000L),
                new LoginEvent("chenyuping", "zujiao", "fail", 5000L),
                new LoginEvent("chenyuping", "koujiao", "fail", 8000L),
                new LoginEvent("chenyuping", "gangmen", "fail", 100000L),
                new LoginEvent("yangdan", "yindao", "fail", 2000L),
                new LoginEvent("yangdan", "gangmen", "fail", 3000L),
                new LoginEvent("yangdan", "gangmen", "success", 4000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                    @Override
                    public long extractTimestamp(LoginEvent loginEvent, long recordTimestamp) {
                        return loginEvent.timestamp;
                    }
                }));

        //2.定义模式，连续三次登录失败
        Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("first")  //第一次登录失败事件
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return loginEvent.eventType.equals("fail");
                    }
                })
                .next("second")  //紧跟着第二次登录失败事件
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return loginEvent.eventType.equals("fail");
                    }
                })
                .next("third")  //紧跟着第三次登录失败事件
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent) throws Exception {
                        return loginEvent.eventType.equals("fail");
                    }
                });

        //3.将模式应用到数据流上，检测复杂事件
        PatternStream<LoginEvent> patternStream = CEP.pattern(loginEventStream.keyBy(event -> event.userId), pattern);

        //4.将检测到的复杂事件提取出来，进行处理得到报警信息
        SingleOutputStreamOperator<String> warningStream = patternStream.select(new PatternSelectFunction<LoginEvent, String>() {
            @Override
            public String select(Map<String, List<LoginEvent>> map) throws Exception {
                //提取复杂事件中的第三次登录失败事件
                LoginEvent firstFailEvent = map.get("first").get(0);
                LoginEvent secondFailEvent = map.get("second").get(0);
                LoginEvent thirdFailEvent = map.get("third").get(0);

                return firstFailEvent.userId + "连续三次登录失败！登录时间："
                        + firstFailEvent.timestamp + ","
                        + secondFailEvent.timestamp + ","
                        + thirdFailEvent.timestamp;
            }
        });

        //打印输出
        warningStream.print();

        env.execute();
    }
}
